package main

import (
	"flag"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s: %s", msg, err)
	}
}

var (
	tag           = flag.String("tag", "", "tag")
	sleepDuration = flag.Duration("sleep", time.Second, "sleep duration")
)

func main() {
	flag.Parse()
	// 连接到RabbitMQ服务器
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	// 创建通道
	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// 声明队列
	q, err := ch.QueueDeclare(
		"hdyy", // 队列名称
		true,   // 持久化
		false,  // 不自动删除
		false,  // 非排他性
		false,  // 不等待
		nil,    // 参数
	)
	failOnError(err, "Failed to declare a queue")

	// 设置每次只获取一条消息，处理完再获取下一条
	err = ch.Qos(
		1,     // 预取数量
		0,     // 预取大小
		false, // 应用于全局
	)
	failOnError(err, "Failed to set QoS")

	// 消费消息
	msgs, err := ch.Consume(
		q.Name, // 队列名称
		*tag,   // 消费者标签
		false,  // 非自动确认消息
		false,  // 非排他性
		false,  // 不接收自己发送的消息
		false,  // 不等待
		nil,    // 参数
	)
	failOnError(err, "Failed to register a consumer")

	// 处理消息
	var forever chan struct{}
	go func() {
		for d := range msgs {
			log.Printf(" [x] Received %s", d.Body)
			// 模拟处理任务
			time.Sleep(*sleepDuration)
			log.Printf(" [x] Done")
			// 确认消息已处理
			err := d.Ack(false)
			if err != nil {
				log.Printf("Failed to ack message: %v", err)
			}
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}
